package com.learn.grpc.nameserver;

import io.grpc.EquivalentAddressGroup;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 模拟注册中心: 服务实例变更后，发送消息通知
 */
@Slf4j
public class NsMockNameCenter {

    // 服务名与实例列表映射
    private static Map<String, List<EquivalentAddressGroup>> serviceMapInstance = new ConcurrentHashMap<>();

    // 监听器列表，当服务实例变更时, 通知 grpc 的监听器
    private List<ServiceInstanceListener> serviceInstanceListeners = new ArrayList<>();

    // 单例模式
    public static final NsMockNameCenter instance = new NsMockNameCenter();

    private NsMockNameCenter() {

        // 固定一个服务名，进行测试
        String serviceName = "rpc-server";

        // 初始化两个实例
        EquivalentAddressGroup server9091 = new EquivalentAddressGroup(Arrays.asList(new InetSocketAddress("localhost", 9091)));
        EquivalentAddressGroup server9092 = new EquivalentAddressGroup(Arrays.asList(new InetSocketAddress("localhost", 9092)));

        AtomicInteger idx = new AtomicInteger(0);

        // 初始化
        serviceMapInstance.put(serviceName, Arrays.asList(server9091));

        // 定时任务: 每隔20s 切换一次服务列表
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleAtFixedRate(()->{
            int left = idx.addAndGet(1) % 2;
            if (left == 0) {
                serviceMapInstance.put(serviceName, Arrays.asList(server9091));
            } else{
                serviceMapInstance.put(serviceName, Arrays.asList(server9092));
            }

            // 变更时打印日志
            List<String> hostPortList = serviceMapInstance.get(serviceName).stream()
                    .map(EquivalentAddressGroup::getAddresses).flatMap(List::stream)
                    .map(item -> (InetSocketAddress) item).map(item -> item.getHostName() + ":" + item.getPort())
                    .collect(Collectors.toList());
            log.info("服务注册实例更新, 服务名称:{}, 实例列表:{}", serviceName, hostPortList);

            // 通知监听器，服务实例发生变更
            for (ServiceInstanceListener serviceInstanceListener : this.serviceInstanceListeners) {
                serviceInstanceListener.updateServiceInstances(serviceName, serviceMapInstance.get(serviceName));
            }
        },5l, 5l, TimeUnit.SECONDS);

    }

    // 注册监听器
    public void registerNameListener(ServiceInstanceListener listener) {
        this.serviceInstanceListeners.add(listener);
    }

    // 通过服务名，获取实例列表
    public static List<EquivalentAddressGroup> getInstanceList(String serviceName) {
        List<EquivalentAddressGroup> serverList  = serviceMapInstance.get(serviceName);
        List<String> addressList = serverList.stream()
                .flatMap(item -> item.getAddresses().stream())
                .map(item -> ((InetSocketAddress) item)
                        .getAddress().getHostAddress() + ":" +  ((InetSocketAddress) item)
                        .getPort())
                .collect(Collectors.toList());
        return serverList;
    }

    // 服务实例监听器
    public interface ServiceInstanceListener {

        // 更细服务实例列表
        public void updateServiceInstances(String serviceName, List<EquivalentAddressGroup> list);

    }
}



